package com.donglink.project.logmonitor.bolt;

import com.donglink.project.logmonitor.domain.MyMessage;
import com.donglink.project.logmonitor.utils.MonitorHandler;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.IBasicBolt;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.IWindowedBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Map;

public class FilterBolt extends BaseRichBolt {

    OutputCollector outputCollector;
    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {
        this.outputCollector=outputCollector;

    }

    @Override
    public void execute(Tuple tuple) {
        //判断是否符合规则，如果符合规则就发往下一个bolt
        String log=tuple.getString(0);
        MyMessage message=MonitorHandler.parseMessage(log);
        //如果这条信息是一条有用信息
        if(message==null)
        {

            return;
        }

        if(MonitorHandler.triggerNotfy(message))
        {
            outputCollector.emit(new Values(message));
        }

        MonitorHandler.tongbu();
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {
        outputFieldsDeclarer.declare(new Fields("filterlog"));
    }
}
